package com.ai.bitcoin.init;

import com.ai.bitcoin.queue.*;
import com.ai.bitcoin.service.helper.NsqServiceHelper;
import com.github.brainlag.nsq.NSQConsumer;
import com.github.brainlag.nsq.lookup.DefaultNSQLookup;
import com.github.brainlag.nsq.lookup.NSQLookup;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
public class NsqService implements InitializingBean, DisposableBean {

    private ExecutorService cachePool;

    private ExecutorService news2bccPool;
    private ExecutorService pricechangePool;
    private ExecutorService anno2bccPool;
    private ExecutorService boomBustPool;
    private ExecutorService buyerPool;
    private ExecutorService ercTwoZeroPool;


    private NSQConsumer newsConsumer;
    private NSQConsumer annoConsumer;
    private NSQConsumer boomBustConsumer;
    private NSQConsumer buyerConsumer;
    private NSQConsumer ercTwoZeroConsumer;


    private MessageQueue newsQueue;
    private MessageQueue annoQueue;
    private MessageQueue boomBustQueue;
    private MessageQueue buerQueue;
    private MessageQueue ercTwoZeroQueue;

    @Autowired
    private NsqServiceHelper nsqServiceHelper;

    @Override
    public void afterPropertiesSet() throws Exception {
        news2bccPool = Executors.newFixedThreadPool(2);
        pricechangePool = Executors.newFixedThreadPool(2);
        anno2bccPool = Executors.newFixedThreadPool(2);
        boomBustPool = Executors.newFixedThreadPool(2);
        buyerPool = Executors.newFixedThreadPool(2);
        ercTwoZeroPool = Executors.newFixedThreadPool(2);


        newsQueue = new MessageQueue(10 * 1000, true);
        annoQueue = new MessageQueue(10 * 1000, true);
        boomBustQueue = new MessageQueue(10 * 1000, true);
        buerQueue = new MessageQueue(10 * 1000, true);
        ercTwoZeroQueue = new MessageQueue(10 * 1000, true);


        NSQLookup lookup = new DefaultNSQLookup();
        lookup.addLookupAddress("47.98.181.191", 4161);

        newsConsumer = new NSQConsumer(lookup, "news2bcc", "news", (message) -> {
            System.out.println("received news: " + new String(message.getMessage()));
            newsQueue.send(new String(message.getMessage()));
            //now mark the message as finished.
            message.finished();

            //or you could requeue it, which indicates a failure and puts it back on the queue.
            //message.requeue();
        });
        newsConsumer.setExecutor(news2bccPool);
        newsConsumer.start();

        annoConsumer = new NSQConsumer(lookup, "anno2bcc", "announce", (message) -> {
            System.out.println("received anno: " + new String(message.getMessage()));
            annoQueue.send(new String(message.getMessage()));
            //now mark the message as finished.
            message.finished();

            //or you could requeue it, which indicates a failure and puts it back on the queue.
            //message.requeue();
        });
        annoConsumer.setExecutor(anno2bccPool);
        annoConsumer.start();

        boomBustConsumer = new NSQConsumer(lookup, "price_change_by_market", "boom_bust_price", (message) -> {
            System.out.println("received boomBust: " + new String(message.getMessage()));
            boomBustQueue.send(new String(message.getMessage()));
            //now mark the message as finished.
            message.finished();

            //or you could requeue it, which indicates a failure and puts it back on the queue.
            //message.requeue();
        });
        boomBustConsumer.setExecutor(boomBustPool);
        boomBustConsumer.start();


        buyerConsumer = new NSQConsumer(lookup, "trading_point", "buyer_seller_point", (message) -> {
            System.out.println("received buyer: " + new String(message.getMessage()));
            buerQueue.send(new String(message.getMessage()));
            //now mark the message as finished.
            message.finished();

            //or you could requeue it, which indicates a failure and puts it back on the queue.
            //message.requeue();
        });
        buyerConsumer.setExecutor(buyerPool);
        buyerConsumer.start();


        ercTwoZeroConsumer = new NSQConsumer(lookup, "erc20_transfer", "erc_two_zero_transfer", (message) -> {
            System.out.println("received ercTwoZero: " + new String(message.getMessage()));
            ercTwoZeroQueue.send(new String(message.getMessage()));
            //now mark the message as finished.
            message.finished();

            //or you could requeue it, which indicates a failure and puts it back on the queue.
            //message.requeue();
        });
        ercTwoZeroConsumer.setExecutor(ercTwoZeroPool);
        ercTwoZeroConsumer.start();


        cachePool = Executors.newCachedThreadPool();
        NewsThread newsThread = new NewsThread(newsQueue, nsqServiceHelper);
        AnnoThread annoThread = new AnnoThread(annoQueue, nsqServiceHelper);
        BoomBustThread boomBustThread = new BoomBustThread(boomBustQueue, nsqServiceHelper);
        BuyerThread buyerThread = new BuyerThread(buerQueue, nsqServiceHelper);
        ErcTwoZeroThread ercTwoZeroThread = new ErcTwoZeroThread(ercTwoZeroQueue, nsqServiceHelper);
        cachePool.submit(newsThread);
        cachePool.submit(annoThread);
        cachePool.submit(boomBustThread);
        cachePool.submit(buyerThread);
        cachePool.submit(ercTwoZeroThread);
    }

    @Override
    public void destroy() throws Exception {
        cachePool.shutdownNow();
        news2bccPool.shutdownNow();
        pricechangePool.shutdownNow();
        cachePool.shutdownNow();
        anno2bccPool.shutdownNow();
        boomBustPool.shutdownNow();
        buyerPool.shutdownNow();
        ercTwoZeroPool.shutdownNow();

        newsConsumer.close();
        annoConsumer.close();
        boomBustConsumer.close();
        buyerConsumer.close();
        ercTwoZeroConsumer.close();
        try {
            //给线程shutdown的时间
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
